package org.apm.data.pusher;

import java.net.InetAddress;
import java.net.UnknownHostException;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.apm.data.util.TimeUtil;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.transport.client.PreBuiltTransportClient;

public class CommonMethod {
    
    public static Client devClient() {
        Settings settings = Settings.builder().put("cluster.name", "Apm-Dev").build();
        TransportClient transportClient = new PreBuiltTransportClient(settings);
        try {
            transportClient.addTransportAddress( new InetSocketTransportAddress(InetAddress.getByName("10.10.102.101"), 9300));
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }
        return transportClient;
    }
    
    public static Client onlineClient() {
        Settings settings = Settings.builder().put("cluster.name", "Apm").build();
        TransportClient transportClient = new PreBuiltTransportClient(settings);
        try {
            transportClient.addTransportAddress( new InetSocketTransportAddress(InetAddress.getByName("10.10.103.102"), 9300));
            transportClient.addTransportAddress( new InetSocketTransportAddress(InetAddress.getByName("10.10.103.112"), 9300));
            transportClient.addTransportAddress( new InetSocketTransportAddress(InetAddress.getByName("10.10.101.107"), 9300));
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }
        return transportClient;
    }
    
    public enum EsBulkProcessor {
        DEV(devClient()),
        ONLINE(onlineClient());

        private Client client = devClient();
        private BulkProcessor esBulkProcessor = null;
        private final Logger log = LogManager.getLogger(EsBulkProcessor.class);
                                                                                                                                       
        // 可调控参数
        private final ByteSizeValue bulkSize = new ByteSizeValue(10, ByteSizeUnit.MB);
        private final TimeValue bulkTime = TimeValue.timeValueSeconds(3);
        private final int correntRequest = 64;
        private final int bulkQuantity = 20000;

        EsBulkProcessor(Client client) {
            this.client = client;
            esBulkProcessorMaker();
        }

        /**
         * params should be set before this method if needed.
         * 
         * @return an bulk processor instance.
         */
        public final BulkProcessor processor() {
            return esBulkProcessor;
        }

        private void esBulkProcessorMaker() {
            esBulkProcessor = BulkProcessor.builder(client, new BulkProcessor.Listener() {
                private int number = 0;
                private int context = 0;
                private long start = 0;

                @Override
                public void beforeBulk(final long executionId, final BulkRequest request) {
                    start = System.currentTimeMillis();
                    context = request.requests().size();
                    number = request.numberOfActions();
                }

                @Override
                public void afterBulk(final long executionId, final BulkRequest request, final BulkResponse response) {
                    long end = System.currentTimeMillis();
                    StringBuilder stringBuilder = new StringBuilder("[Bulk][").append(executionId).append("] 始于: ")
                            .append(TimeUtil.timestampToDate(start)).append(", 终于: ").append(TimeUtil.timestampToDate(end))
                            .append(", took: ").append(response.getTookInMillis()).append(", ").append(number).append(", ")
                            .append(context).append(", ").append(response.buildFailureMessage());
                    log.info(stringBuilder.toString());
                }

                @Override
                public void afterBulk(final long executionId, final BulkRequest request, final Throwable failure) {
                    StringBuilder stringBuilder = new StringBuilder("[Bulk][").append(executionId).append("] ")
                            .append(failure.getMessage());
                    log.error(stringBuilder);
                }
            }).setBulkActions(bulkQuantity).setBulkSize(bulkSize).setFlushInterval(bulkTime)
                    .setConcurrentRequests(correntRequest).build();

        }
    }
}
